using System;
using System.Collections;
using System.Threading;

namespace Sedna.Core.Messaging.InMemoryImpl
{
	public class InMemoryBroadcastTopic : IProvider
	{

		private Queue queue = Queue.Synchronized(new Queue());
		private IEndPoint destination;

		public IEndPoint Destination
		{
			get { return destination; }
			set { destination = value; }
		}

		public IMessageSession GetSession()
		{
			return new Session(this);
		}

		private void Put(IMessage msg)
		{
			lock(queue.SyncRoot)
			{
				queue.Enqueue(msg);
				Monitor.PulseAll(queue.SyncRoot);
			}
		}

		private IMessage Get()
		{
			lock(queue.SyncRoot)
			{
				while(queue.Count == 0)
				{
					try
					{
						Monitor.Wait(queue.SyncRoot);
					}
					catch(ThreadAbortException e)
					{
						Thread.CurrentThread.Interrupt();
						return null;
					}
				}
				return (IMessage)queue.Dequeue();
			}
		}


		public class Session : IMessageSession
		{

			private InMemoryBroadcastTopic topic;
			private static IEndPoint anonymousEndPoint = new AnonymousEndPoint();

			public Session(InMemoryBroadcastTopic topic)
			{
				this.topic = topic;
			}

			public void Send(object payload)
			{
				Send(topic.destination, payload);
			}

			public void Send(IEndPoint endPoint, object payload)
			{
				if(endPoint != topic.destination)
				{
					throw new NotSupportedException("Alternative destinations are not supported");
				}
				IMessage message = new SimpleMessage();
				message.Source = anonymousEndPoint;
				message.Destination = topic.destination;
				message.Payload = payload;
				Send(message);
			}

			public void Send(IMessage message)
			{
				if(message.Destination != null && message.Destination != topic.destination)
				{
					throw new NotSupportedException("Alternative destinations are not supported");
				}
				message.Destination = topic.destination;
				topic.Put(message);
			}

			public IMessage Receive()
			{
				return (IMessage)topic.Get();
			}

			public IMessage Receive(int timeout)
			{
				throw new NotImplementedException();
			}
		}
	}
}
